package com.njtoyo.taxi.admin.kafka.handler;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.convert.Convert;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.json.JSONUtil;
import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.njtoyo.taxi.admin.cache.PrimaryCachePutter;
import com.njtoyo.taxi.admin.handler.common.CustomerCouponHandler;
import com.njtoyo.taxi.admin.handler.common.DriverRewardHandler;
import com.njtoyo.taxi.admin.kafka.model.ActivityOrderCompletionRewardsModel;
import com.njtoyo.taxi.admin.mapper.master.entity.*;
import com.njtoyo.taxi.admin.mapper.master.enums.RecommendationPromoteChannel;
import com.njtoyo.taxi.admin.mapper.master.enums.RideOrderState;
import com.njtoyo.taxi.admin.mapper.master.mapper.*;
import com.njtoyo.taxi.admin.rest.pojo.dto.activity.orderCompletionReward.CustomerDetailSpecification;
import com.njtoyo.taxi.admin.rest.pojo.dto.activity.orderCompletionReward.DriverDetailSpecification;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;

import java.math.BigDecimal;
import java.util.List;

import com.njtoyo.taxi.admin.kafka.topic.ActivityTopic;

/**
 * 完单奖励处理
 */
@Slf4j
@Component
public class ActivityOrderCompletionRewards {
    @Autowired
    private UserMapper userMapper;
    @Autowired
    private CustomerRegionActivityRuleMapper customerRegionActivityRuleMapper;
    @Autowired
    private DriverRegionActivityRuleMapper driverRegionActivityRuleMapper;
    @Autowired
    private ActivityRewardLogMapper activityRewardLogMapper;
    @Autowired
    private CustomerCouponHandler customerCouponHandler;
    @Autowired
    private RideOrderMapper rideOrderMapper;
    @Autowired
    private RecommendationMapper recommendationMapper;
    @Autowired
    private DriverMapper driverMapper;
    @Autowired
    private DriverRewardHandler driverRewardHandler;
    @Autowired
    private PrimaryCachePutter primaryCachePutter;

    @KafkaListener(topics = {ActivityTopic.Order_Completion_Rewards})
    public void activityOrderCompletionRewards(List<ConsumerRecord<String, String>> recordList, Acknowledgment ack) {
        log.info("=========================开始处理主题【{}】消息, 消息数量: {}=========================", ActivityTopic.Order_Completion_Rewards, recordList.size());

        if (CollUtil.isEmpty(recordList)) {
            log.error("---主题【{}】消息内容为空----", ActivityTopic.Order_Completion_Rewards);
            ack.acknowledge();
            return;
        }

        for (ConsumerRecord<String, String> record : recordList) {
            try {
                log.info("开始处理消息, key: {}, offset: {}, partition: {}", record.key(), record.offset(), record.partition());

                processSingleRecord(record);

                log.debug("消息处理完成, key: {}, offset: {}, partition: {}", record.key(), record.offset(), record.partition());
            } catch (Exception e) {
                log.error("处理订单完成奖励消息失败, 消息key: {}, 错误: {}", record.key(), e.getMessage(), e);
                // 可以选择是否ack，这里选择ack避免消息堆积
            }
        }

        // 不管是否有异常，最后确认消费
        ack.acknowledge();
        log.info("=========================完成处理主题【{}】所有消息=========================", ActivityTopic.Order_Completion_Rewards);
    }

    /**
     * 处理单条订单完成奖励消息
     */
    private void processSingleRecord(ConsumerRecord<String, String> record) {
        ActivityOrderCompletionRewardsModel model = JSON.parseObject(record.value(), ActivityOrderCompletionRewardsModel.class);
        log.info("处理订单完成奖励消息, customerId: {}, couponId: {}", model.getCustomerId(), model.getCouponId());

        User user = userMapper.selectById(model.getCustomerId());
        if (ObjectUtil.isNull(user)) {
            log.info("未找到用户, customerId: {}", model.getCustomerId());
            return;
        }

        log.info("找到用户信息, userId: {}, customerRegionId: {}", user.getId(), user.getCustomerRegionId());

        // 处理乘客奖励
        processPassengerReward(user, model);

        // 处理司机奖励
        processDriverReward(model);
    }

    /**
     * 处理乘客奖励逻辑
     */
    private void processPassengerReward(User user, ActivityOrderCompletionRewardsModel model) {
        log.info("开始处理乘客奖励, userId: {}, customerRegionId: {}", user.getId(), user.getCustomerRegionId());

        CustomerRegionActivityRule customerRegionActivityRule = customerRegionActivityRuleMapper.selectOne(Wrappers.<CustomerRegionActivityRule>lambdaQuery()
                .eq(CustomerRegionActivityRule::getCustomerRegionId, user.getCustomerRegionId())
                .eq(CustomerRegionActivityRule::getIsOpen, true));

        if (ObjectUtil.isNull(customerRegionActivityRule)) {
            log.info("未找到乘客完单活动规则, customerRegionId: {}", user.getCustomerRegionId());
            return;
        }

        if (ObjectUtil.isNull(customerRegionActivityRule) || !customerRegionActivityRule.getIsOpen()) {
            log.info("乘客完单活动规则不存在或未开启, customerRegionId: {}", user.getCustomerRegionId());
            return;
        }

        log.info("找到乘客活动规则, 规则ID: {}, 活动状态: {}", customerRegionActivityRule.getId(), customerRegionActivityRule.getIsOpen());

        // 检查用户是否在活动时间范围内注册
        if (!DateUtil.isIn(user.getCreatedAt(), customerRegionActivityRule.getStartedAt(), customerRegionActivityRule.getEndedAt())) {
            log.info("用户注册时间不在活动范围内, 用户ID: {}, 注册时间: {}, 活动开始: {}, 活动结束: {}", user.getId(), user.getCreatedAt(), customerRegionActivityRule.getStartedAt(), customerRegionActivityRule.getEndedAt());
            return;
        }

        log.info("用户注册时间在活动范围内, 用户ID: {}, 注册时间: {}", user.getId(), DateUtil.formatDateTime(user.getCreatedAt()));

        if (ObjectUtil.isNull(model.getCouponId())) {
            log.info("乘客: 未使用优惠券 不发新的优惠券");
            return;
        }

        CustomerDetailSpecification customerDetailSpecification = customerRegionActivityRule.parseDetailSpecification();
        log.info("新客完单奖励, 封顶金额: {}, 奖励数量: {}", customerDetailSpecification.getTotalAmount(), customerDetailSpecification.getCount());

        // 查找乘客已发放奖励
        List<ActivityRewardLog> grants = activityRewardLogMapper.selectList(Wrappers.<ActivityRewardLog>lambdaQuery()
                .eq(ActivityRewardLog::getGrantedId, user.getId())
                .eq(ActivityRewardLog::getActivityCategory, 0));

        log.info("用户已发放奖励数量: {}", grants.size());

        // 查找某个活动已发放金额
        BigDecimal grantAmount = activityRewardLogMapper.grantAmountBy(customerRegionActivityRule.getId(), 0);
        log.info("活动已发放金额: {}, 活动总封顶金额: {}", grantAmount, customerDetailSpecification.getTotalAmount());


        if (customerDetailSpecification.getTotalAmount().compareTo(grantAmount.add(customerDetailSpecification.getCoupon().getValue())) <= 0) {
            log.info("已发放奖励金额已达上限, 当前已发放: {}, 总上限: {}", grantAmount, customerDetailSpecification.getTotalAmount());
            return;
        }

        if (grants.size() >= customerDetailSpecification.getCount()) {
            log.info("已发放奖励数量已达上限, 当前已发放: {}, 总上限: {}", grants.size(), customerDetailSpecification.getCount());
            return;
        }

        // 发放奖励
        log.info("开始发放乘客奖励, 规则ID: {}, 用户ID: {}, 客户区域ID: {}", customerRegionActivityRule.getId(), user.getId(), user.getCustomerRegionId());

        customerCouponHandler.sendCustomerCoupon(customerRegionActivityRule, user.getId(), user.getCustomerRegionId(), DateUtil.beginOfDay(DateUtil.offsetDay(DateUtil.date(), 1)));

        // 记录奖励日志
        ActivityRewardLog addLog = new ActivityRewardLog();
        addLog.setGrantedId(user.getId());
        addLog.setRewardId(model.getCouponId());
        addLog.setRewardValue(Convert.toBigDecimal(customerDetailSpecification.getCoupon().getValue()));
        addLog.setRewardCategory(0);
        addLog.setRefId(customerRegionActivityRule.getId());
        addLog.setActivityCategory(0);
        addLog.setCustomerRegionId(user.getCustomerRegionId());

        activityRewardLogMapper.insert(addLog);
        log.info("乘客奖励发放成功, 奖励日志ID: {}", addLog.getId());
    }

    /**
     * 处理司机奖励逻辑
     */
    private void processDriverReward(ActivityOrderCompletionRewardsModel model) {
        log.info("开始处理司机奖励 {}");

        // 检查是否首单
        List<RideOrder> rideOrders = rideOrderMapper.selectList(Wrappers.<RideOrder>lambdaQuery()
                .eq(RideOrder::getUserId, model.getCustomerId())
                .eq(RideOrder::getState, RideOrderState.done)
                .not(i -> i.eq(RideOrder::getRideId, model.getRideId())));

        if (CollUtil.isNotEmpty(rideOrders)) {
            log.info("乘客不是首单, 已有完成订单数量: {}, 参数信息：{}", rideOrders.size(), JSONUtil.toJsonPrettyStr(model));
            return;
        }

        log.info("乘客是首单, 开始检查推荐关系");

        // 查询推荐关系
        Recommendation recommendation = recommendationMapper.selectOne(Wrappers.<Recommendation>lambdaQuery()
                .eq(Recommendation::getUserId, model.getCustomerId())
                .eq(Recommendation::getIsValid, true)
                .eq(Recommendation::getPromoteChannel, RecommendationPromoteChannel.driver));

        if (ObjectUtil.isNull(recommendation)) {
            log.info("未找到正确的推荐关系, 参数信息：{}", JSONUtil.toJsonPrettyStr(model));
            return;
        }

        log.info("找到推荐关系, 推荐人ID: {}", recommendation.getFromId());

        Driver driver = driverMapper.selectById(recommendation.getFromId());
        if (ObjectUtil.isNull(driver)) {
            log.info("司机不存在, 推荐人ID: {}", recommendation.getFromId());
            return;
        }
        primaryCachePutter.deleteDriverCacheById(driver.getId());
        log.info("找到司机信息, 司机ID: {}, 司机区域ID: {}", driver.getId(), driver.getDriverRegionId());

        // 查询司机活动规则
        DriverRegionActivityRule driverRegionActivityRule = driverRegionActivityRuleMapper.selectOne(Wrappers.<DriverRegionActivityRule>lambdaQuery()
                .eq(DriverRegionActivityRule::getDriverRegionId, driver.getDriverRegionId())
                .eq(DriverRegionActivityRule::getIsOpen, true));

        if (ObjectUtil.isNull(driverRegionActivityRule)) {
            log.info("未找到司机完单活动规则, 司机区域ID: {}", driver.getDriverRegionId());
            return;
        }

        log.info("找到司机活动规则, 规则ID: {}, 活动状态: {}", driverRegionActivityRule.getId(), driverRegionActivityRule.getIsOpen());

        // 检查活动时间范围
        if (!DateUtil.isIn(DateUtil.date(), driverRegionActivityRule.getStartedAt(), driverRegionActivityRule.getEndedAt())) {
            log.info("司机完单活动规则未开始或者已结束, 开始时间: {}, 结束时间: {}, 当前时间: {}", driverRegionActivityRule.getStartedAt(), driverRegionActivityRule.getEndedAt(), DateUtil.date());
            return;
        }

        log.info("司机活动在有效时间内");

        // 解析司机奖励规则
        DriverDetailSpecification driverDetailSpecification = driverRegionActivityRule.parseDetailSpecification();
        BigDecimal totalAmount = driverDetailSpecification.getTotalAmount();
        log.info("司机奖励总封顶金额: {}", totalAmount);

        // 检查已发放金额
        BigDecimal driverGrantAmount = activityRewardLogMapper.grantAmountBy(driverRegionActivityRule.getId(), 1);
        log.info("司机已发放金额: {}, 总封顶金额: {}", driverGrantAmount, totalAmount);

        if (totalAmount.compareTo(driverGrantAmount) <= 0) {
            log.info("司机已发放金额超出阈值, 停止后续奖励操作, 已发放: {}, 总上限: {}", driverGrantAmount, totalAmount);
            return;
        }

        // 发放司机奖励
        log.info("开始发放司机奖励, 司机ID: {}, 奖励金额: {}", driver.getId(), driverDetailSpecification.getReward().getValue());

        Long wareId = driverRewardHandler.sendDriverReward(driver.getId(), driverDetailSpecification.getReward().getValue());

        // 记录司机奖励日志
        ActivityRewardLog addLog = new ActivityRewardLog();
        addLog.setGrantedId(driver.getId());
        addLog.setRewardId(wareId);
        addLog.setRewardValue(driverDetailSpecification.getReward().getValue());
        addLog.setRewardCategory(1);
        addLog.setRefId(driverRegionActivityRule.getId());
        addLog.setActivityCategory(1);
        addLog.setDriverRegionId(driver.getDriverRegionId());
        activityRewardLogMapper.insert(addLog);
        log.info("司机奖励发放成功");
    }
}